package com.zhaosc.spark.stream

import org.apache.spark.streaming.StreamingContext
import org.apache.spark.SparkConf
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.kafka.KafkaManager
import com.zhaosc.spark.constant.SysConst
import org.apache.spark.streaming.kafka.OffsetRange
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.HasOffsetRanges
import org.apache.spark.rdd.RDD

object KafkaManagerDemo {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("KafkaManagerDemo").set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val ssc = new StreamingContext(conf, Seconds(5)) //5秒处理一次消息

    val kafkaParams = Map[String, String](
      "metadata.broker.list" -> SysConst.KAFA_METADATA_BROKER_LIST,
      "group.id" -> SysConst.KAFA_PRS_USERFACE_GROUP_ID,
      "fetch.message.max.bytes" -> String.valueOf(5 * 1024 * 1024))
    val topicsSet = SysConst.KAFA_PRS_USERFACE_TOPIC.split(",").toSet
    val km = new KafkaManager(kafkaParams)
    val messages = km.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

    var offsetRanges = Array[OffsetRange]()
    val linesRDD = messages.transform { rdd =>
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
    }.flatMap(_._2.split("\n"))
    var flag = true
    if (flag) {
      linesRDD.foreachRDD ( (rdd, time) =>
        {
          println("-------------------------------------------")
          println(s"Time: $time")
          processRdd(rdd);
          //rdd.saveAsTextFile("D:\\tmp\\kafka")
          println("-------------------------------------------")
          km.commitOffsetsToZK(offsetRanges)
        }
      )

      flag = false
    }

    ssc.start()
    ssc.awaitTermination()

  }

  def processRdd(rdd: RDD[String]): Unit = {
    rdd.foreach(println(_))
  }

}